package study.mq.kafka.consumer.listener._3_batch;

import cn.hutool.core.thread.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.stereotype.Component;

/**
 * 批量拉取
 */
@Slf4j
@Component
public class Batch_02_Listener {

    //    @KafkaListener(topics = "batch-topic")
    public void onMessage(ConsumerRecords<String, String> records) {
        log.info(">>>>>> count: {}", records.count());

        String topic = "";
        Iterable<ConsumerRecord<String, String>> myRecords = records.records(topic);
        myRecords.forEach(record -> {
            ThreadUtil.sleep(1 * 1000);
            //
            int partition = record.partition();
            long offset = record.offset();
            log.info("[{}, {}] offset= {}", topic, partition, offset);
        });
    }
}
